package com.swak.job.worker;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.OS;
import com.swak.job.Job;
import com.swak.job.JobConfig;
import com.swak.job.JobConfig.Config;
import com.swak.job.JobConstants;
import com.swak.job.enums.Role;
import com.swak.job.queue.QueueService;
import com.swak.loadbalance.LoadBalances;
import com.swak.metrics.MetricsFactory;
import com.swak.metrics.NamedMetrics;
import com.swak.reactivex.threads.Context;
import com.swak.reactivex.threads.Contexts;
import com.swak.registry.RegistryService;
import com.swak.registry.URL;
import com.swak.utils.Maps;
import com.swak.utils.StringUtils;
import com.swak.vertx.invoker.GenericFluxInvoker;
import com.swak.vertx.transport.script.Script;

/**
 * 启动服务注册，和发现
 * 
 * @author lifeng
 * @date 2021年2月25日 下午2:32:15
 */
@SuppressWarnings("unchecked")
public abstract class AbstractWorker implements Worker, JobConstants {

	protected Logger logger = LoggerFactory.getLogger(Worker.class);

	protected final QueueService queueService;
	protected final RegistryService registryService;
	protected final MetricsFactory metricsFactory;
	protected final Job job;
	protected final Role role;
	protected final Object service;
	protected final Method method;
	protected final GenericFluxInvoker invoker;
	protected final AtomicBoolean state = new AtomicBoolean(false);
	protected final NamedMetrics<Object> namedMetrics;

	protected JobConfig jobConfig;
	protected Config config;
	protected URL oldUrl;
	protected URL url;
	protected volatile boolean opened;

	// 调度、执行 ： 都有队列
	protected String queue;

	public AbstractWorker(JobConfig config, Role role, Job job, 
			QueueService queueService, RegistryService registryService, MetricsFactory metricsFactory) {
		this.job = job;
		this.role = role;
		this.queueService = queueService;
		this.registryService = registryService;
		this.metricsFactory = metricsFactory;
		this.service = this.job.getJobService();
		this.method = this.role == Role.Dispatcher ? job.getDispatchMethod() : job.getExecuteMethod();
		this.invoker = new GenericFluxInvoker(this.service);

		// 刷新参数
		this.refresh(config);

		// Job.Module.WechatJob.Dispatcher
		this.namedMetrics = this.metricsFactory != null ? (NamedMetrics<Object>) this.metricsFactory
				.createNamedMetrics("Job" + Constants.MODULE_SEPARATE + this.jobConfig.getModule()
						+ Constants.MODULE_SEPARATE + this.job.getId() + Constants.MODULE_SEPARATE + this.role.name())
				: null;
	}

	/**
	 * 配置刷新
	 */
	@Override
	public void refresh(JobConfig config) {
		this.jobConfig = config;
		this.config = this.jobConfig.getConfigs().get(job.getId());
		this.opened = this.getRefreshAbleOpened();
		this.oldUrl = this.url;
		this.url = this.initUrl(null);
		this.queue = this.queueService.parseQueue(this.role, this.url);
		this.doRefresh();
		this.reStart();
	}

	/**
	 * 重新启动： 已启动的才需要重新启动
	 */
	protected void reStart() {
		if (this.state.get()) {
			this.stop();
			this.start();
		}
	}

	/**
	 * 执行刷新操作
	 */
	protected void doRefresh() {
	}

	/**
	 * 初始化 URL 信息
	 * 
	 * @return
	 */
	protected URL initUrl(Map<String, String> specificParameters) {
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Application_KEY, this.jobConfig.getApplicationName());
		parameters.put(Constants.Module_Key, this.jobConfig.getModule());
		parameters.put(Constants.Service_Key, this.job.getId());
		parameters.put(Constants.Category_Key, this.role.name());
		parameters.put(Group_key, this.job.getGroup());
		parameters.put(Worker_key, this.service.getClass().getName());
		parameters.put(Isolate_key, String.valueOf(this.getRefreshAbleIsolate()));
		parameters.put(Weight_key, String.valueOf(this.getRefreshAbleWeight()));
		parameters.put(Open_key, String.valueOf(this.getRefreshAbleOpened()));
		if (specificParameters != null && !specificParameters.isEmpty()) {
			parameters.putAll(specificParameters);
		}
		return new URL(Constants.Registry_Job_Key, OS.ip(), 0, Constants.Registry_Job_Worker_Key, parameters);
	}

	/****    可刷新的参数  ****/
	protected Integer getRefreshAbleWeight() {
		if (this.config != null && this.config.getWeight() != null) {
			return this.config.getWeight();
		}
		return this.job.getWeight();
	}

	protected Boolean getRefreshAbleOpened() {
		if (this.config != null && this.config.getOpened() != null) {
			return this.config.getOpened();
		}
		return this.job.isOpened();
	}

	protected Boolean getRefreshAbleIsolate() {
		if (this.config != null && this.config.getIsolateExecute() != null) {
			return this.config.getIsolateExecute();
		}
		return this.job.isIsolate();
	}

	protected String getRefreshAbleCronExpression() {
		if (this.config != null && StringUtils.isNoneBlank(this.config.getCronExpression())) {
			return this.config.getCronExpression();
		}
		return this.job.getCronExpression();
	}

	protected Boolean getRefreshAbleOrder() {
		if (this.config != null && this.config.getOrder() != null) {
			return this.config.getOrder();
		}
		return this.job.getOrder();
	}
	
	protected Boolean getRefreshAbleAutorun() {
		if (this.config != null && this.config.getAutorun() != null) {
			return this.config.getAutorun();
		}
		return this.job.getAutorun();
	}

	protected LoadBalances getRefreshAbleLoadBalances() {
		if (this.config != null && this.config.getLoadBalance() != null) {
			return this.config.getLoadBalance();
		}
		return this.job.getDispatchLoadBalance();
	}

	protected Integer getRefreshAbleRetryTimes() {
		if (this.config != null && this.config.getRetryTimes() != null) {
			return this.config.getRetryTimes();
		}
		return this.job.getRetryTimes();
	}

	protected Integer getRefreshAbleParallel() {
		if (this.config != null && this.config.getParallel() != null) {
			return this.config.getParallel();
		}
		return this.job.getParallel();
	}

	/**
	 * 执行相关方法
	 * 
	 * @throws Throwable
	 */
	protected Object invoke(Object[] args) throws Throwable {
		Context context = Contexts.currentContext();
		Object result = this.invoker.doInvoke(this.method, args);
		if (result != null && result instanceof CompletionStage && context != null) {
			CompletionStage<Object> resultFuture = (CompletionStage<Object>) result;
			CompletableFuture<Object> future = new CompletableFuture<>();
			resultFuture.whenComplete((v, e) -> {
				context.runOnContext(() -> {
					if (e != null) {
						future.completeExceptionally(e);
					} else {
						future.complete(v);
					}
				});
			});
			return future;
		}
		return result;
	}

	/**
	 * 执行相关方法
	 * 
	 * @throws Throwable
	 */
	protected Object invoke(Script script) throws Throwable {
		Context context = Contexts.currentContext();
		Object result = this.invoker.doInvoke(script);
		if (result != null && result instanceof CompletionStage && context != null) {
			CompletionStage<Object> resultFuture = (CompletionStage<Object>) result;
			CompletableFuture<Object> future = new CompletableFuture<>();
			resultFuture.whenComplete((v, e) -> {
				context.runOnContext(() -> {
					if (e != null) {
						future.completeExceptionally(e);
					} else {
						future.complete(v);
					}
				});
			});
			return future;
		}
		return result;
	}

	@Override
	public URL getURL() {
		return url;
	}

	@Override
	public void start() {
		if (state.get()) {
			return;
		}
		this.registryService.register(this.url);
		this.state.set(true);
	}

	@Override
	public void stop() {
		if (!state.get()) {
			return;
		}
		URL unregisterURL = this.oldUrl != null ? this.oldUrl : this.url;
		this.registryService.unregister(unregisterURL);
		this.state.set(false);
	}
}